package com.wsx.demo;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.FloatWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

public class Test {

	private static final String INPUT_PATH = "hdfs://localhost:9000/user/root/input/";
	private static final String OUTPUT_PATH = "hdfs://localhost:9000/user/root/output/";

	
	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException, URISyntaxException {
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf, "test");
		
		FileSystem fs = null;
		fs = FileSystem.get(new URI(OUTPUT_PATH), conf);
		if(fs.exists(new Path(OUTPUT_PATH))){
			fs.delete(new Path(OUTPUT_PATH), true);
		}
		fs.close();
		
		fs = FileSystem.get(new URI(INPUT_PATH), conf);
		if(fs.exists(new Path(INPUT_PATH))){
			fs.delete(new Path(INPUT_PATH), true);
		}
		fs.close();
		
		fs = FileSystem.get(conf);
		Path src = new Path("doc/simple/");
		Path dst = new Path(INPUT_PATH);
		fs.copyFromLocalFile(src, dst);
		fs.close();
		
		job.setJarByClass(Test.class);
		
		job.setMapperClass(TestMapper.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);
		
		job.setReducerClass(TestReducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(FloatWritable.class);
		
		job.setInputFormatClass(TextInputFormat.class);
		job.setOutputFormatClass(TextOutputFormat.class);
		
		FileInputFormat.setInputPaths(job, new Path(INPUT_PATH));
		FileOutputFormat.setOutputPath(job, new Path(OUTPUT_PATH));
	
		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}

}
